package com.cssrc.wenjie.hbase.util;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.List;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.cssrc.wenjie.hbase.annotation.HBaseColumn;
import com.cssrc.wenjie.hbase.annotation.HBaseRowKey;
import com.cssrc.wenjie.hbase.annotation.HBaseTable;

/**   
* @des Hbase tool
* @author wenjie   
* @date 2018年8月31日 下午6:45:49 
* @version V1.0  
*/
public class HBaseUtil {
    public static Logger log = Logger.getLogger(HBaseUtil.class);
    
    public static <T> T result2Bean(Result result, T obj) throws Exception {
        Assert.notNull(result, "Result不能为null！");
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            String value = null;
            field.setAccessible(true);
            if (field.isAnnotationPresent(HBaseRowKey.class)) {
                value = new String(result.getRow());
            } else if (field.isAnnotationPresent(HBaseColumn.class)) {
                HBaseColumn hBaseColumn = field.getAnnotation(HBaseColumn.class);
                String family = hBaseColumn.family();
                String qualifier = hBaseColumn.qualifier();
                boolean timeStamp = hBaseColumn.timestamp();
                if (StringUtils.isEmpty(family) || StringUtils.isEmpty(qualifier)) {
                    continue;
                }
                if (!timeStamp) {
                    value = new String(result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier)));
                }
                List<Cell> cells = result.getColumnCells(Bytes.toBytes(family), Bytes.toBytes(qualifier));
                if (cells.size() > 0) {
                    Cell cell = cells.get(0);
                    value = Bytes.toString(cell.getValue()) + "";
                } else {
                    value = "";
                }
            }
            if (value != null) {
                String fieldName = field.getName();
                String firstLetter = fieldName.substring(0, 1).toUpperCase();
                String setMethodName = "set" + firstLetter + fieldName.substring(1);
                Method setMethod = clazz.getMethod(setMethodName, new Class[] {field.getType()});
                setMethod.invoke(obj, new Object[] {value});
            }
        }
        return obj;
    }
    
    public static Put bean2Put(Object obj) throws Exception {
        byte[] rowKey = getRowKey(obj);
        Put put = new Put(rowKey);
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            if (!field.isAnnotationPresent(HBaseColumn.class)) {
                continue;
            }
            HBaseColumn hBaseColumn = field.getAnnotation(HBaseColumn.class);
            String family = hBaseColumn.family();
            String qualifier = hBaseColumn.qualifier();
            if (StringUtils.isEmpty(family) || StringUtils.isEmpty(qualifier)) {
                continue;
            }
            field.setAccessible(true);
            Object fieldObj = field.get(obj);
            if (field.get(obj) != null || !StringUtils.isEmpty(field.get(obj))) {
                put.addImmutable(Bytes.toBytes(family),
                    Bytes.toBytes(qualifier),
                    Bytes.toBytes(field.get(obj).toString()));
            }
        }
        return put;
    }
    
    private static byte[] getRowKey(Object obj) {
        byte[] rowKey = null;
        Class<?> clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        try {
            for (Field field : fields) {
                field.setAccessible(true);
                if (field.isAnnotationPresent(HBaseRowKey.class)) {
                    rowKey = Bytes.toBytes(field.get(obj).toString());
                    break;
                }
            }
            if (rowKey == null) {
                Assert.notNull(rowKey, clazz.getName() + "类的方法中必须有HBaseRowKey注解！");
            }
        } catch (Exception e) {
            log.error("", e);
        }
        return rowKey;
    }
    
    public static String getTableName(Object obj) {
        String tableName = "";
        try {
            HBaseTable hBaseTable = obj.getClass().getAnnotation(HBaseTable.class);
            tableName = hBaseTable.tableName();
        } catch (Exception e) {
            log.error(obj.getClass() + "没有添加HBaseTable注解！", e);
        }
        return tableName;
    }
}
